// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package kafka

import (
	"fmt"
	"gitee.com/go-wares/log/base"
	"gitee.com/go-wares/log/conf"
	"github.com/IBM/sarama"
	"sync"
	"time"
)

// Adapter
// is a component that print log message on terminal.
type Adapter struct {
	bucket    *base.Bucket[*base.Item]
	formatter base.LoggerFormatter
	kc        *sarama.Config
	mu        *sync.Mutex
	producer  sarama.SyncProducer
}

// NewAdapter
// creates a new adapter for terminal.
func NewAdapter() *Adapter {
	return (&Adapter{
		formatter: &Formatter{},
	}).init()
}

// GetBucket
// return logger items bucket.
func (o *Adapter) GetBucket() *base.Bucket[*base.Item] {
	return o.bucket
}

// GetFormatter
// returns the logger formatter.
func (o *Adapter) GetFormatter() base.LoggerFormatter {
	return o.formatter
}

// Log
// add logger item into bucket.
func (o *Adapter) Log(item *base.Item) {
	o.bucket.Add(item)
}

// SetFormatter
// set the logger formatter.
func (o *Adapter) SetFormatter(formatter base.LoggerFormatter) {
	o.formatter = formatter
}

// +---------------------------------------------------------------------------+
// | Access methods                                                            |
// +---------------------------------------------------------------------------+

func (o *Adapter) getKafkaConfig() *sarama.Config {
	if o.kc == nil {
		c := conf.Get()
		o.kc = sarama.NewConfig()

		o.kc.Net.DialTimeout = time.Duration(c.KafkaAdapter.DialTimeout) * time.Second
		o.kc.Net.ReadTimeout = time.Duration(c.KafkaAdapter.ReadTimeout) * time.Second
		o.kc.Net.WriteTimeout = time.Duration(c.KafkaAdapter.WriteTimeout) * time.Second

		o.kc.Producer.RequiredAcks = sarama.NoResponse
		o.kc.Producer.Timeout = time.Duration(c.KafkaAdapter.WriteTimeout) * time.Second
		o.kc.Producer.Retry.Max = 1
		o.kc.Producer.Retry.Backoff = 300 * time.Millisecond
		o.kc.Producer.Return.Errors = true
		o.kc.Producer.Return.Successes = true
		o.kc.Producer.CompressionLevel = sarama.CompressionLevelDefault
	}
	return o.kc
}

func (o *Adapter) getProducer() (sarama.SyncProducer, error) {
	var err error

	o.mu.Lock()
	defer o.mu.Unlock()

	// Create kafka producer if not set.
	if o.producer == nil {
		var (
			c  = conf.Get()
			kc = o.getKafkaConfig()
			p  sarama.SyncProducer
		)

		// Kafka producer creator.
		if p, err = sarama.NewSyncProducer(c.KafkaAdapter.Addresses, kc); err == nil {
			o.producer = p
		}
	}
	return o.producer, err
}

// Init
// adapter fields.
func (o *Adapter) init() *Adapter {
	o.mu = &sync.Mutex{}
	o.bucket = base.NewBucket[*base.Item]("kafka-logger", o.run)
	o.bucket.Start()
	return o
}

// Run
// publish log message to kafka.
func (o *Adapter) run(items []*base.Item) {
	var (
		c   = conf.Get()
		err error
		ms  = make([]*sarama.ProducerMessage, 0)
		p   sarama.SyncProducer
	)

	// Release
	// log items when publish completed.
	defer func() {
		for _, item := range items {
			item.Release()
		}
	}()

	// Range items
	// as message list.
	for _, item := range items {
		ms = append(ms, &sarama.ProducerMessage{
			Topic:     c.KafkaAdapter.Topic,
			Value:     sarama.StringEncoder(fmt.Sprintf(`%s`, o.formatter.Format(item))),
			Timestamp: item.Time,
		})
	}

	// Do nothing
	// if log items count is empty.
	if len(ms) == 0 {
		return
	}

	// Get producer and send messages to kafka.
	if p, err = o.getProducer(); err == nil {
		err = p.SendMessages(ms)
	}
	if err != nil {
		println("send messages fail:", err.Error())
	}
}
